package com.itcast.spark

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * DESC:整合SparkSQL和SparkStreaming---->使用SparkSQL的DSL和SQL完成对应数据的查询
 * 使用两者结合的方法实现wordcount
 * 1-准备环境StreamingContext，内部调用了SparkCOntext，指定数据处理时间Seconds(5)
 * 2-准备Connector数据源数据读入
 * 3-对于当前的数据需要进行flatMap扁平化
 * 4-结合SparkSQL实现的WordCount
 * 5-ssc.start 开启流式应用
 * 6-ssc.awaitTermination等待程序异常终止
 * 7-ssc.stop(true,true)
 */
object SparkSQLAndSparkStreamingBase {
  def main(args: Array[String]): Unit = {
    //1-准备环境StreamingContext，内部调用了SparkCOntext，指定数据处理时间Seconds(5)
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQLAndSparkStreamingBase")
    val sc = new SparkContext(config = conf)
    val ssc = new StreamingContext(sc, Seconds(5))
    sc.setLogLevel("WARN")

    //2-准备Connector数据源数据读入
    val receiveData: ReceiverInputDStream[String] = ssc.socketTextStream("node01", 9999, StorageLevel.MEMORY_AND_DISK)
    //3-对于当前的数据需要进行flatMap扁平化
    val data: DStream[String] = receiveData.flatMap(_.split(" "))
    //4-结合SparkSQL实现的WordCount
    data.foreachRDD(rdd => {
      val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
      import spark.implicits._
      val dataDF: DataFrame = rdd.toDF("words")
      //这是第一种方法
      // val wordcountResult: Dataset[Row] = dataDF.groupBy("words").count().orderBy($"count".desc)
      dataDF.createOrReplaceTempView("worsText")
      //这是第2种方法
      val result: DataFrame = spark.sql("select words,count(words) as count from worsText group by words order by count desc")
      result.show()
    })
    //5-ssc.start 开启流式应用
    ssc.start()
    //6-ssc.awaitTermination等待程序异常终止
    ssc.awaitTermination()
    //7-ssc.stop(true,true)
    ssc.stop(true, true)
  }
}
